You could also drop this transformer task once BEEFY initialized and worker starts, the worker consumes finality notifications as they come in so no block pinning issues from that point on.
Not really important, but saves us an extra task/future that's running forever and copies data from pinned->unpinned for no functional reason.

That would be an option, true. However I think the impact of this additional future is pretty slim, since we are just using intra-task concurrency here. If we were to go for this route, we would need to re-initialize the transformer in case of a consensus reset where we start the loop again from the beginning. Also we would need to do some kind of "hand-over" for the streams to make absolutely sure that we don't miss any finality notifications. Since this raises the complexity I opted for the most simple solution here.

Would it work if we just did something like:

let mut finality_notifications = client
		.finality_notification_stream()
		.map(UnpinnedFinalityNotification::from)
		.fuse();

?

And in the function definitions we could use:

		finality_notifications: &mut Fuse<
			impl Stream<Item = UnpinnedFinalityNotification<B>> + Unpin,
		>

Would it work if we just did something like:

let mut finality_notifications = client
		.finality_notification_stream()
		.map(UnpinnedFinalityNotification::from)
		.fuse();

?

This would map the stream, but not solve our problem. The mapping takes place when the next item is retrieved from the stream. Our problem is that next() is not called while we are waiting for the header ancestry to become available. We need an extra future that runs concurrently for this mapping to happen in the background.

Oh, I see, you're right. This makes sense. But, at a first glance, personally I think it would be better if instead we had a wrapper over the finality notifications stream with some internal state. Something like:

FinalityStream {
    inner: original finality stream,
    mode: Limited | Full
    queue: queued notifications while in Limited mode
}

We could have it in Limited mode while waiting for the header ancestry to become available, and then switching it to Full. In Limited mode we can just take the notifications, transform them to unpinned and push them to a queue. We still need to pump it while waiting for async_initialize() to finish, just as you do with the transformer here.

Also maybe we could drop all the finality notifications, except for the ones for mandatory headers while in Limited mode ?

Anyway, not a blocker on my side. If the current approach works, I'm ok with merging it, and then I can experiment with this wrapper.

I mean we could do that, but I am not sure if there is any benefit. This transformer is already super simple and gets the job done. I think we can merge the current approach and then iterate on it, for now my priority is to not pin unnecessary blocks. Right now I am warp syncing some nodes to make sure nothing unexpected happens.

Also maybe we could drop all the finality notifications, except for the ones for mandatory headers while in
Limited mode ?

I was experimenting with this and think its worth following up on this idea. However there are some fields that we set based on the finality notifications (even non-mandatory) and since I am just getting familiar with the code, I opted for the simplest-yet-working approach.

Yes we can follow-up with micro-optimizations like removing the extra async task in the middle once initialization is complete. For now, this is good for fixing the issue.

PR just needs prdoc and is good to go from my point of view.

Will add that as soon as my nodes synced correctly. Then we can merge 👍